package com.atguigu.day12;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class Flink01_SQL_Demo {
    public static void main(String[] args) throws Exception {

/*  方式一：
        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        //获取表的环境
        TableEnvironment tableEnv = TableEnvironment.create(environmentSettings);*/

        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        //2.通过流的执行环境获取表的环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //3.创建表 连接到Kafka
        tableEnv.executeSql("CREATE TABLE source_kafka( \n" +
                "id String, \n" +
                "ts bigint , \n" +
                "vc int)\n" +
                "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "  'properties.group.id' = 'atguigu',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'topic' = 'source_topic',\n" +
                "  'value.format' = 'csv'\n" +
                ")");

        Table table = tableEnv.sqlQuery("select * from source_kafka");


        //第一种，获取到Table对象然后进行操作
   /*     TableResult tableResult = table.execute();
        tableResult.print();*/

        //方式二：直接获取TableResult对象

        TableResult tableResult = tableEnv.executeSql("select * from source_kafka");

        tableResult.print();

/*
        //TODO  方式三： 将Table对象转为流 进行操作
        DataStream<WaterSensor> waterSensorDataStream = tableEnv.toDataStream(table, WaterSensor.class);

        waterSensorDataStream.print();

        env.execute();
*/
//        env.execute();

    }
}
